package com.dayu.tcp.service;

import com.dayu.tcp.conf.K1TcpConf;
import com.dayu.tcp.count.CountInfo;
import com.dayu.tcp.exception.SocketRuntimeException;
import com.dayu.tcp.future.InvokeFuture;
import com.dayu.tcp.protocol.K1HeartbeatMsg;
import com.dayu.tcp.protocol.K1RestfulRequest;
import com.dayu.tcp.protocol.K1RestfulResponse;
import com.dayu.tcp.service.server.Server;
import com.dayu.tcp.service.server.ServerContext;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.*;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
import lombok.Getter;
import lombok.NonNull;
import lombok.Setter;

import java.net.SocketAddress;
import java.nio.channels.ClosedChannelException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;

/**
 * @author yehangtech
 * @date 2023/11/26 22:00
 * @description 相当于通道
 */
public class WrappedChannel implements Channel {

    @Getter
    private final Channel channel;

    public WrappedChannel(@NonNull Channel channel) {
        this.channel = channel;
    }

    /**
     * 缺省的AEX通道加密密钥
     */
    @Setter
    @Getter
    private String encryptKey = K1TcpConf.ENCRYPT_KEY;

    /**
     * 会话相关信息缓存
     */
    @Getter
    private final ConcurrentMap<String, Object> sessionCache = new ConcurrentHashMap<>();

    /**
     * 异步转同步
     * key->sequenceId
     */
    @Getter
    private final ConcurrentMap<Long, InvokeFuture> futureMap = new ConcurrentHashMap<>();

    private final ChannelFutureListener sendSuccessListener = future -> {
        if (future.isSuccess()) {
            Server server = ServerContext.getServerContext().getServer();
            if (server != null) {
                CountInfo countInfo = server.getCountInfo();
                countInfo.getSendNum().incrementAndGet();
                countInfo.setLastSend(System.currentTimeMillis());
            }
        }
    };

    /**
     * 异步发送消息，原生netty
     *
     * @param message
     * @return
     */
    public ChannelFuture send(Object message) {
        return this.writeAndFlush(message);
    }

    /**
     * 同步发送消息，指定超时时间
     *
     * @param message 发送的消息体
     * @param timeout timeout<=0 则代表同步一直等待
     * @return
     */
    public K1RestfulResponse sendSync(final K1RestfulRequest message, long timeout, TimeUnit timeUnit) {
        final InvokeFuture invokeFuture = new InvokeFuture();
        try {
            invokeFuture.setChannel(this);
            //存储InvokeFuture
            futureMap.put(message.getSequence(), invokeFuture);
            //发送Request对象
            ChannelFuture channelFuture = writeAndFlush(message);
            channelFuture.addListener((ChannelFutureListener) future -> {
                if (!future.isSuccess()) {
                    futureMap.remove(message.getSequence());
                    invokeFuture.setCause(future.cause());
                }
            });
        } catch (Throwable ex) {
            throw new SocketRuntimeException(ex);
        }

        Object retObject;
        //设置超时时间
        if (timeout > 0) {
            retObject = invokeFuture.getResult(timeout, timeUnit);
        } else {
            //一直等待，直到Response返回
            retObject = invokeFuture.getResult();
        }
        return (K1RestfulResponse) retObject;
    }

    @Override
    public ChannelId id() {
        return channel.id();
    }

    @Override
    public EventLoop eventLoop() {
        return channel.eventLoop();
    }

    @Override
    public Channel parent() {
        return channel.parent();
    }

    @Override
    public ChannelConfig config() {
        return channel.config();
    }

    @Override
    public boolean isOpen() {
        return channel.isOpen();
    }

    @Override
    public boolean isRegistered() {
        return channel.isRegistered();
    }

    @Override
    public boolean isActive() {
        return channel.isActive();
    }

    @Override
    public ChannelMetadata metadata() {
        return channel.metadata();
    }

    @Override
    public SocketAddress localAddress() {
        return channel.localAddress();
    }

    @Override
    public SocketAddress remoteAddress() {
        return channel.remoteAddress();
    }

    @Override
    public ChannelFuture closeFuture() {
        return channel.closeFuture();
    }

    @Override
    public boolean isWritable() {
        return channel.isWritable();
    }

    @Override
    public long bytesBeforeUnwritable() {
        return channel.bytesBeforeUnwritable();
    }

    @Override
    public long bytesBeforeWritable() {
        return channel.bytesBeforeWritable();
    }

    @Override
    public Unsafe unsafe() {
        return channel.unsafe();
    }

    @Override
    public ChannelPipeline pipeline() {
        return channel.pipeline();
    }

    @Override
    public ByteBufAllocator alloc() {
        return channel.alloc();
    }

    @Override
    public ChannelFuture bind(SocketAddress socketAddress) {
        return channel.bind(socketAddress);
    }

    @Override
    public ChannelFuture connect(SocketAddress socketAddress) {
        return channel.connect(socketAddress);
    }

    @Override
    public ChannelFuture connect(SocketAddress socketAddress, SocketAddress socketAddress1) {
        return channel.connect(socketAddress, socketAddress1);
    }

    @Override
    public ChannelFuture disconnect() {
        return channel.disconnect();
    }

    @Override
    public ChannelFuture close() {
        return this.innerClose();
    }

    private ChannelFuture innerClose() {
        ChannelFuture channelFuture = channel.close();
        //cancel所有等待中的InvokeFuture
        for (InvokeFuture invokeFuture : futureMap.values()) {
            if (!invokeFuture.isDone()) {
                invokeFuture.setCause(new ClosedChannelException());
            }
        }
        return channelFuture;
    }

    @Override
    public ChannelFuture deregister() {
        return channel.deregister();
    }

    @Override
    public ChannelFuture bind(SocketAddress socketAddress, ChannelPromise channelPromise) {
        return channel.bind(socketAddress, channelPromise);
    }

    @Override
    public ChannelFuture connect(SocketAddress socketAddress, ChannelPromise channelPromise) {
        return channel.connect(socketAddress, channelPromise);
    }

    @Override
    public ChannelFuture connect(SocketAddress socketAddress, SocketAddress socketAddress1, ChannelPromise channelPromise) {
        return channel.connect(socketAddress, socketAddress1, channelPromise);
    }

    @Override
    public ChannelFuture disconnect(ChannelPromise channelPromise) {
        return channel.disconnect(channelPromise);
    }

    @Override
    public ChannelFuture close(ChannelPromise channelPromise) {
        return channel.close(channelPromise);
    }


    @Override
    public ChannelFuture deregister(ChannelPromise channelPromise) {
        return channel.deregister(channelPromise);
    }

    @Override
    public Channel read() {
        return channel.read();
    }

    @Override
    public ChannelFuture write(Object message) {
        return this.write(message, true);
    }

    public ChannelFuture write(Object message, boolean isStatistic) {
        ChannelFuture future = channel.write(message);
        if (isStatistic) {
            if (!(message instanceof K1HeartbeatMsg)) {
                future.addListener(sendSuccessListener);
            }
        }
        return future;
    }

    @Override
    public ChannelFuture write(Object message, ChannelPromise channelPromise) {
        return this.write(message, channelPromise, true);
    }

    public ChannelFuture write(Object message, ChannelPromise channelPromise, boolean isStatistic) {
        ChannelFuture future = channel.write(message, channelPromise);
        if (isStatistic) {
            if (!(message instanceof K1HeartbeatMsg)) {
                future.addListener(sendSuccessListener);
            }
        }
        return future;
    }

    @Override
    public Channel flush() {
        return channel.flush();
    }

    @Override
    public ChannelFuture writeAndFlush(Object message, ChannelPromise channelPromise) {
        return this.writeAndFlush(message, channelPromise, true);
    }

    public ChannelFuture writeAndFlush(Object message, ChannelPromise channelPromise, boolean isStatistic) {
        ChannelFuture future = channel.writeAndFlush(message, channelPromise);
        if (isStatistic) {
            if (!(message instanceof K1HeartbeatMsg)) {
                future.addListener(sendSuccessListener);
            }
        }
        return future;
    }

    @Override
    public ChannelFuture writeAndFlush(Object message) {
        return this.writeAndFlush(message, true);
    }

    /**
     * 客户端调用服务端的写入方法
     * @param message
     * @param isStatistic
     * @return
     */
    public ChannelFuture writeAndFlush(Object message, boolean isStatistic) {
        ChannelFuture future = channel.writeAndFlush(message);
        if (isStatistic) {
            if (!(message instanceof K1HeartbeatMsg)) {
                future.addListener(sendSuccessListener);
            }
        }
        return future;
    }

    @Override
    public ChannelPromise newPromise() {
        return channel.newPromise();
    }

    @Override
    public ChannelProgressivePromise newProgressivePromise() {
        return channel.newProgressivePromise();
    }

    @Override
    public ChannelFuture newSucceededFuture() {
        return channel.newSucceededFuture();
    }

    @Override
    public ChannelFuture newFailedFuture(Throwable throwable) {
        return channel.newFailedFuture(throwable);
    }

    @Override
    public ChannelPromise voidPromise() {
        return channel.voidPromise();
    }

    @Override
    public <T> Attribute<T> attr(AttributeKey<T> key) {
        return channel.attr(key);
    }

    @Override
    public <T> boolean hasAttr(AttributeKey<T> key) {
        return channel.hasAttr(key);
    }

    @Override
    public int compareTo(Channel o) {
        return channel.compareTo(o);
    }
}
